package com.whmall.service.crm.module.supplierreview.impl;

import com.whmall.component.entity.dto.PageReqDTO;
import com.whmall.component.entity.dto.PageRspDTO;
import com.whmall.component.enumeration.OperatorTypeEnum;
import com.whmall.component.util.FlatListUtils;
import com.whmall.service.crm.module.supplierreview.SupplierReviewQueryCriteria;
import com.whmall.service.crm.module.supplierreview.SupplierReviewRepository;
import com.whmall.service.crm.module.supplierreview.entity.SupplierReviewDO;
import com.whmall.service.crm.module.supplierreview.entity.SupplierReviewTagStatisticsDO;
import com.whmall.service.crm.module.supplierreview.impl.dao.SupplierReviewMapper;
import com.whmall.service.crm.module.supplierreview.impl.dao.SupplierReviewTagStatisticsMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 供应商评论统计同步器
 *
 * @author alex
 * @since 2021/1/19
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class SupplierReviewTagStatisticsSynchronizer {
    private final SupplierReviewRepository supplierReviewRepository;
    private final SupplierReviewMapper supplierReviewMapper;
    private final SupplierReviewTagStatisticsMapper supplierReviewTagStatisticsMapper;
    private final PlatformTransactionManager platformTransactionManager;
    private TransactionTemplate transactionTemplate;

    @PostConstruct
    public void init() {
        this.transactionTemplate = new TransactionTemplate(platformTransactionManager);
    }

    public void sync() {
        log.info("开始同步所有未统计的评论...");
        final long startTime = System.currentTimeMillis();

        //
        // 分页请求，每次处理当前时间之前的100条评论数据，以ID排序
        PageReqDTO pageRequest = PageReqDTO.builder()
                .pageNum(1)
                .pageSize(100)
                .sortFields(Collections.singletonList("id"))
                .build();

        SupplierReviewQueryCriteria criteria = new SupplierReviewQueryCriteria();
        criteria.setNotCount(true);
        criteria.setCreateBefore(new Date());

        AtomicInteger handleCount = new AtomicInteger();

        for (; ; ) {
            PageRspDTO<SupplierReviewDO> page = supplierReviewRepository.findReviewPageByQueryCriteria(pageRequest, criteria);

            //
            // 进行统计
            syncReviews(page.getRecords());
            handleCount.addAndGet(page.getRecords().size());

            //
            // 跳出，当分页的列表数量小于请求的数量
            if (page.getRecords().size() < pageRequest.getPageSize()) {
                break;
            }

            //
            // 下一页
            pageRequest.setPageNum(pageRequest.getPageNum() + 1);
        }

        log.info("同步完成，一共处理了 {} 条评论，耗时 {}s",
                handleCount,
                (System.currentTimeMillis() - startTime) / 1000f);
    }

    public void incrementCompanyTags(Long companyId, List<Integer> tagIds, int delta) {
        if (null == tagIds || tagIds.isEmpty()) {
            return;
        }

        List<SupplierReviewTagStatisticsDO> statsList = supplierReviewTagStatisticsMapper.findByCompanyIdAndTagIn(companyId, tagIds);
        if (statsList.size() != tagIds.size()) {
            throw new IllegalArgumentException("传入的标签ID统计与已经存在的统计数据不匹配");
        }

        for (SupplierReviewTagStatisticsDO stats : statsList) {
            SupplierReviewTagStatisticsDO update = new SupplierReviewTagStatisticsDO();
            update.setId(stats.getId());
            update.setTagCount1(stats.getTagCount1() + delta);
            update.setUpdateTime(new Date());
            update.setBeforeUpdateTimeAsVersion(stats.getUpdateTime());
            int affectRows = supplierReviewTagStatisticsMapper.optimisticUpdateByIdSelective(update);
            if (affectRows != 1) {
                throw new IllegalArgumentException("更新失败，请稍后再试");
            }
        }
    }

    private void syncReviews(List<SupplierReviewDO> reviews) {
        if (null == reviews || reviews.isEmpty()) {
            return;
        }

        //
        // 确保在事务内处理
        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
            @Override
            protected void doInTransactionWithoutResult(TransactionStatus status) {
                doSync(reviews);
            }
        });
    }

    private void doSync(List<SupplierReviewDO> reviews) {
        Map<Long, List<SupplierReviewDO>> companyReviewsMap = reviews
                .stream()
                .collect(Collectors.groupingBy(SupplierReviewDO::getCompanyId));

        for (Long companyId : companyReviewsMap.keySet()) {
            List<SupplierReviewDO> uncountReviews = companyReviewsMap.get(companyId);
            Map<Integer, Long> countingTagMap = uncountReviews
                    .stream()
                    .flatMap(e -> FlatListUtils.toList(e.getTagList(), Integer::valueOf).stream())
                    .collect(Collectors.groupingBy(e -> e, Collectors.counting()));
            if (countingTagMap.isEmpty()) {
                continue;
            }
            Map<Integer, SupplierReviewTagStatisticsDO> companyTagMap = supplierReviewTagStatisticsMapper.findByCompanyIdAndTagIn(companyId, countingTagMap.keySet())
                    .stream()
                    .collect(Collectors.toMap(SupplierReviewTagStatisticsDO::getTag, e -> e));

            countingTagMap.forEach((tag, count) -> {
                SupplierReviewTagStatisticsDO stats = companyTagMap.get(tag);
                if (null == stats) {
                    //
                    // 第一次统计该公司该标签
                    SupplierReviewTagStatisticsDO insert = new SupplierReviewTagStatisticsDO();
                    insert.setCompanyId(companyId);
                    insert.setCreateTime(new Date());
                    insert.setTag(tag);
                    insert.setTagCount1(count.intValue());
                    supplierReviewTagStatisticsMapper.insertSelective(insert);
                } else {
                    //
                    // 累加统计
                    SupplierReviewTagStatisticsDO update = new SupplierReviewTagStatisticsDO();
                    update.setId(stats.getId());
                    update.setTagCount1(stats.getTagCount1() + count.intValue());
                    update.setUpdateTime(new Date());
                    update.setBeforeUpdateTimeAsVersion(stats.getUpdateTime());
                    int affectRows = supplierReviewTagStatisticsMapper.optimisticUpdateByIdSelective(update);
                    if (affectRows != 1) {
                        log.warn("尝试更新标签统计 {} 的数据时发生了错误，该数据跟读取时已经不一致，为避免数据不一致，终止整个任务，尝试下一次继续", stats.getId());
                        throw new IllegalArgumentException("乐观更新失败，回滚数据，尝试下一次继续统计");
                    }
                }
            });
        }

        //
        // 更新评论为已统计
        for (SupplierReviewDO review : reviews) {
            int affectRows = supplierReviewMapper.optimisticUpdateAsCounted(review);
            if (affectRows != 1) {
                log.warn("尝试将评论 {} 标记为已统计时发生了错误，该数据跟读取时已经不一致，为避免数据不一致，终止整个任务，尝试下一次继续", review.getId());
                throw new IllegalArgumentException("乐观更新失败，回滚数据，尝试下一次继续统计");
            }
        }
    }

}
